package com.huatai.datacenter.service;

import com.huatai.datacenter.entity.ClusterEntity;
import com.huatai.datacenter.mapper.ClusterMapper;
import com.huatai.datacenter.utils.KafkaAdmins;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.clients.admin.DescribeLogDirsResult;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.requests.DescribeLogDirsResponse;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.apache.kafka.common.Node;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * @author Lion
 * @date 2023/4/13  10:56
 */
@Service
@Slf4j
public class KafkaAdminService {

	@Autowired
	ClusterMapper clusterInfoMapper;

	private AdminClient adminClient;

	private Map<String, KafkaAdmins> cacheKafkaMap  =  new HashMap<>( 1 << 4) ;

	private Map<String, AdminClient> cacheAdminClient = new HashMap<>(1 << 4) ;


	/**
	 * @author Lion
	 * @description kafka健康监测
	 * @param brokerAddr
	 * @return boolean
	 */
	public boolean kafkaIsHeath(String brokerAddr) {
		boolean flag = false;
		KafkaAdmins admin = null;
		try {
			Properties props = new Properties();
			props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerAddr);
			props.put("request.timeout.ms", 1000);
			props.put("connections.max.idle.ms", 1000);
			admin = new KafkaAdmins(props);

			Collection<Node> nodes = admin.describeCluster()
				.nodes()
				.get();

			if(admin!=null && nodes.size() > 0) {
				flag = true;
			}
		} catch (Exception e) {
			log.warn("connect kafka error.",e);
		}finally {
			try {
				if(admin!=null) {
					admin.close();
				}
			} catch (IOException e) {}
		}
		return flag;

	}

	/**
	 * @author Lion
	 * @description 获取kafka
	 * @param clusterId
	 * @return com.huatai.datacenter.utils.KafkaAdmins
	 */
	public KafkaAdmins getKafkaAdmins (String clusterId)  {
		synchronized (cacheKafkaMap) {
			ClusterEntity cluster =  clusterInfoMapper.selectById(clusterId);
			if(!cacheKafkaMap.containsKey(clusterId)) {
				Properties props = new Properties();
				props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, cluster.getBroker());
				this.cacheKafkaMap.put(clusterId, new KafkaAdmins(props));
			}
			return  this.cacheKafkaMap.get(clusterId);
		}
	}


	/**
	 * 获取adminClient，kafka监控使用
	 *
	 * @param clusterId 集群ID
	 * @return 该集群对应的AdminClient
	 * @author lizhong
	 * @date 2023年8月10日
	 */
	public AdminClient getAdminClientByClusterID(String clusterId) {
		synchronized (cacheAdminClient) {
			ClusterEntity cluster = clusterInfoMapper.selectById(clusterId);
			if (!cacheAdminClient.containsKey(clusterId)) {
				Properties props = new Properties();
				props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, cluster.getBroker());
				this.cacheAdminClient.put(clusterId, AdminClient.create(props));
			}
			return this.cacheAdminClient.get(clusterId);
		}
	}

}
